7. kafka序列化&反序列化

2023-10-25 00:37| 来源: 网络整理| 查看: 265

序列化

kafka序列化消息是在生产端，序列化后，消息才能网络传输。而构造KafkaProducer代码如下：

Properties props = new Properties(); props.put("bootstrap.servers", "10.0.55.229:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); kafkaProducer = new KafkaProducer(props);

属性key.serializer和value.serializer就是key和value指定的序列化方式。无论是key还是value序列化和反序列化实现都是一样的，所以接下来都只以value的序列化和反序列为例。

StringSerializer

StringSerializer是内置的字符串序列化方式，核心源码如下：

/** * String encoding defaults to UTF8 and can be customized by setting the property key.serializer.encoding, * value.serializer.encoding or serializer.encoding. The first two take precedence over the last. */ public class StringSerializer implements Serializer { private String encoding = "UTF8"; ... ... @Override public byte[] serialize(String topic, String data) { try { // 如果数据为空，那么直接返回null即可 if (data == null) return null; else // 否则将String序列化，即转为byte[]即可 return data.getBytes(encoding); } catch (UnsupportedEncodingException e) { throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + encoding); } } @Override public void close() { // nothing to do } } 自定义序列化

和内置的StringSerializer字符串序列化一样，如果要自定义序列化方式，需要实现接口Serializer。假设每个字段按照下图所示的方式自定义序列化：自定义序列化格式

下面是一个简单的自定义实现（假设自定义Order类型的Serializer，Order类中有一些String，Integer，Long，Date类型的属性–其他类型暂不支持，读者可以自行扩展）：

/** * @author wangzhenfei9 * @version 1.0.0 * @since 2018年06月22日 */ public class OrderSerializer implements Serializer { @Override public void configure(Map configs, boolean isKey) { // nothing to do } @Override pub

【本文地址】

公司简介

联系我们